import org.apache.flink.api.java.DataSet;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import org.apache.flink.api.*;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.*;

import java.util.Arrays;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

public class SyntaxTry
{


    public static void main(String[] args) throws Exception
    {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        DataStream<Order> orderA = env.fromCollection(Arrays.asList(
                new Order(1L, "beer", 3),
                new Order(3L, "rubber", 2),
                new Order(1L, "diaper", 4)
        ));

//-----------------------------------------------------------------------------------


//        下面3种写法完全等效(任选1种即可)
//        tEnv.registerDataStream("orderA", orderA, "user,product,amount");

//        tEnv.createTemporaryView("orderA", orderA, "user,product,amount");

        Table tableA = tEnv.fromDataStream(orderA, "user,product,amount");
        tEnv.registerTable("orderA", tableA);

//-----------------------------------------------------------------------------------

//        下面3种写法完全等效(任选1种即可)
//        tEnv.registerFunction("hashcode", new HashCode(10));

//        tEnv.createTemporaryFunction("hashcode", new HashCode(10));
        tEnv.createTemporarySystemFunction("hashcode", new HashCode(10));

//-----------------------------------------------------------------------------------

//        下面5种写法完全等效(任选1种即可)
//        Table result = tEnv.from("orderA").select("product,hashcode(product)");
//        tEnv.toAppendStream(result, Row.class).print();
//        env.execute();



//        Table result=tEnv.sqlQuery("select product,hashcode(product) from orderA");
//        tEnv.toAppendStream(result,Row.class).print();
//        env.execute();


       tableA.select($("product"),call("hashcode","product")).execute().print();




//        Table result=tEnv.from("orderA").select($("product"),call("hashcode","product"));
//        tEnv.toAppendStream(result, Row.class).print();
//        env.execute();



    }
}
